package com.xyw.code.bus.center.event;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.UUID;

/**
 * Created with IntelliJ IDEA.
 * User: xuyiwei
 * Date: 2020/2/18
 * Time: 下午9:21
 * Email: 1328312923@qq.com
 * Description: 发送消息并异步监听 ack
 **/
@Component
@Slf4j
public class MessageSendAsync implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    private RabbitTemplate rabbitTemplate;

    /**
     * 通过构造函数注入 RabbitTemplate 依赖
     *
     * @param rabbitTemplate
     */
    @Autowired
    public MessageSendAsync(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        // 设置消息到达 exchange 时，要回调的方法，每个 RabbitTemplate 只支持一个 ConfirmCallback
        rabbitTemplate.setConfirmCallback(this);
        // 设置消息无法到达 queue 时，要回调的方法
        rabbitTemplate.setReturnCallback(this);
    }

    /**
     * 发送消息
     *
     * @param exchange   交换机
     * @param routingKey 路由建
     * @param message    消息实体
     * @param correlationData 唯一id对象
     */
    public void sendMsg(String exchange, String routingKey, Object message,CorrelationData correlationData) {
        // 构造包含消息的唯一id的对象，id 必须在该 channel 中始终唯一
        if(correlationData==null || StringUtils.isBlank(correlationData.getId())){
            correlationData = new CorrelationData(UUID.randomUUID().toString());
        }
        log.info("ID为: {}", correlationData.getId());
        // todo 先将 message 入库，在将 message 的数据库ID 、message的消息id message的初始状态(发送中)等信息入库

        // 完成 数据落库，消息状态打标后，就可以安心发送 message
        rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
    }


    /**
     * 如果消息到达 exchange, 则 confirm 回调, ack = true
     * 如果消息不到达 exchange, 则 confirm 回调, ack = false
     * 需要设置spring.rabbitmq.publisher-confirms=true
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {

        if (ack) {
            log.info("消息投递成功,ID为: {}", correlationData.getId());
            // todo 操作数据库，将 correlationId 这条消息状态改为投递成功
            return;
        }
        log.error("消息投递失败,ID为: {},错误信息: {}", correlationData.getId(), cause);
        // todo 操作数据库，将 correlationId 这条消息状态改为投递失败
    }

    /**
     * exchange 到达 queue, 则 returnedMessage 不回调
     * exchange 到达 queue 失败, 则 returnedMessage 回调
     * 需要设置spring.rabbitmq.publisher-returns=true
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        // correlationId 就是发消息时设置的 id
        String correlationId = message.getMessageProperties().getHeaders().get("spring_returned_message_correlation").toString();
        log.error("没有找到对应队列，消息投递失败,ID为: {}, replyCode {} , replyText {}, exchange {} routingKey {}",
                correlationId, replyCode, replyText, exchange, routingKey);
        // todo 操作数据库，将 correlationId 这条消息状态改为投递失败
    }

    /**
     * 通用处理错误消息的方法
     * @param message
     * @param channel
     */
    public void dealFailMessage(Message message, Channel channel) {
        // redelivered = true, 表明该消息是重复处理消息
        Boolean redelivered = message.getMessageProperties().getRedelivered();
        /**
         * 这里对消息重入队列做设置，例如将消息序列化缓存至 Redis, 并记录重入队列次数
         * 如果该消息重入队列次数达到一次次数，比如3次，将不再重入队列，直接拒绝
         * 这时候需要对消息做补偿机制处理
         *
         * channel.basicNack与channel.basicReject要结合越来使用
         *
         */
        try {
            if (redelivered) {
                /**
                 * 1. 对于重复处理的队列消息做补偿机制处理
                 * 2. 从队列中移除该消息，防止队列阻塞
                 */
                // 消息已重复处理失败, 扔掉消息
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息
                log.error("消息[{}]重新处理失败，扔掉消息", new String(message.getBody()));
            }
            // redelivered != true,表明该消息是第一次消费
            if (!redelivered) {
                // 消息重新放回队列
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                log.error("消息[{}]处理失败，重新放回队列", new String(message.getBody()));
            }

        } catch (IOException e1) {
            e1.printStackTrace();
        }
    }
}
